package com.afdor.rws.event.driven.channel.rocketmq.consumer.topic;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.exception.BusinessException;
import com.afdor.rws.event.driven.api.EventHolder;
import com.afdor.rws.event.driven.api.Serialization;
import com.afdor.rws.event.driven.api.channel.consumer.EventConsumer;
import com.afdor.rws.event.driven.api.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.rocketmq.config.RocketMqEventConsumerConfig;
import com.afdor.rws.event.driven.channel.rocketmq.utils.RocketMqUtils;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.support.channel.consumer.AbstractEventConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * RocketMq消费者实现
 *
 * @author 悭梵
 * @date Created in 2018-12-28 14:59
 */
@Slf4j
@Extension(value = "rocketMqTopic", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventConsumer.class)
public class RocketMqEventConsumer<EH extends EventHolder, CC extends RocketMqEventConsumerConfig> extends AbstractEventConsumer<EH, CC> implements MessageListenerConcurrently, MessageListenerOrderly {

    /**
     * Rocket MQ 消费者
     */
    private volatile DefaultMQPushConsumer consumer;

    @Override
    protected void doStart() {
        shutdownRocketMQPushConsumer();

        try {
            CC rocketMqEventConsumerConfig = getConfig();
            // TODO 此处注意分片
            this.consumer = new DefaultMQPushConsumer(RocketMqUtils.getTopicSubQueueName(getEventSourceName(), getConsumerName()));
            this.consumer.setNamesrvAddr(rocketMqEventConsumerConfig.getNamesrvAddr());
            // 程序第一次启动从消息队列头获取数据,如果非第一次启动，那么按照上次消费的位置继续消费
            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 设置订阅模式
            this.consumer.setMessageModel(MessageModel.CLUSTERING);
            // 通过修改Consumer的consumerThreadMin和consumerThreadMax来设置线程数
            // 设置最小线程数
            this.consumer.setConsumeThreadMin(NumberUtils.max(1, NumberUtils.min(rocketMqEventConsumerConfig.getConsumerThreadMin(), rocketMqEventConsumerConfig.getDefaultTopicQueueNums())));
            // 设置最大线程数
            this.consumer.setConsumeThreadMax(NumberUtils.max(1, this.consumer.getConsumeThreadMin(), rocketMqEventConsumerConfig.getConsumerThreadMax()));
            // 订阅指定主题下多有Tag类型的消息,都订阅消息
            this.consumer.subscribe(RocketMqUtils.getTopicName(getEventSourceName()), "*");
            // 可以修改每次消费消息的数量，默认设置是每次消费1条
            this.consumer.setConsumeMessageBatchMaxSize(NumberUtils.max(1, rocketMqEventConsumerConfig.getBatchSize()));
            // 注册消费的监听
            if (rocketMqEventConsumerConfig.isOrderlyConsumer()) {
                this.consumer.registerMessageListener((MessageListenerOrderly) this);
            } else {
                this.consumer.registerMessageListener((MessageListenerConcurrently) this);
            }
            this.consumer.start();
        } catch (Exception e) {
            stop();
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            }
            throw new BusinessException("初始化启动消费者失败", e);
        }
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageList, ConsumeConcurrentlyContext context) {
        Event event = null;
        EventHandler eventHandler = getEventHandler();
        Serialization serialization = getSerialization();
        for (MessageExt message : messageList) {
            try {
                event = serialization.unserialize(new String(message.getBody()));
                eventHandler.handler(event);
            } catch (Exception e) {
                // 重试次数限制
                if (message.getReconsumeTimes() >= getConfig().getMaxRetryCount()) {
                    log.error("处理事件异常，并已达到最大重试次数限制", e);
                    // TODO 该条消息可采用其他方式存储与消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                log.warn("处理事件异常，稍后重试", e);
                // 重试
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageList, ConsumeOrderlyContext context) {
        // 设置自动提交
        if (!context.isAutoCommit()) {
            context.setAutoCommit(true);
        }
        Event event = null;
        for (MessageExt message : messageList) {
            try {
                event = getSerialization().unserialize(new String(message.getBody()));
                getEventHandler().handler(event);
            } catch (Exception e) {
                // 重试次数限制
                if (message.getReconsumeTimes() >= getConfig().getMaxRetryCount()) {
                    log.error("处理事件异常，并已达到最大重试次数限制", e);
                    // TODO 该条消息可采用其他方式存储与消费
                    return ConsumeOrderlyStatus.SUCCESS;
                }
                log.warn("处理事件异常，稍后重试", e);
                // 重试
                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
            }
        }

        return ConsumeOrderlyStatus.SUCCESS;
    }

    @Override
    protected void doStop() {
        shutdownRocketMQPushConsumer();
    }

    /**
     * 关闭Rocket MQ的消费者
     */
    private void shutdownRocketMQPushConsumer() {
        if (this.consumer != null) {
            this.consumer.shutdown();
            this.consumer = null;
        }
    }


}
